
package hasthi.container.p2p;

import static hasthi.common.constants.HasthiConstants.*;
import hasthi.common.AbstractMessageBag;
import hasthi.common.BaseContext;
import hasthi.common.ComponetFactory;
import hasthi.common.ErrorManager;
import hasthi.common.HasthiException;
import hasthi.common.SafeScheduler;
import hasthi.common.Utils;
import hasthi.common.XmlBeanUtils;
import hasthi.common.HasthiException.FaultCode;
import hasthi.common.constants.HasthiConstants;
import hasthi.common.constants.WSDMConstants;
import hasthi.container.CommunicationContianier;
import hasthi.container.MessageContext;
import hasthi.container.MessageProcessor;
import hasthi.container.SimpleMessageContext;
import hasthi.container.UnicastContianier;
import hasthi.coordinator.ManageMeRequest;
import hasthi.coordinator.RemoteManagerContext;
import hasthi.manager.AssignResourcesDocument;
import hasthi.manager.CoordinatorHeartbeatDocument;
import hasthi.manager.CoordinatorStepDownNoticeDocument;
import hasthi.manager.ElectedNoticeDocument;
import hasthi.manager.GetCoordinatorDocument;
import hasthi.manager.GetCoordinatorResponseDocument;
import hasthi.manager.GetLiveManagersDocument;
import hasthi.manager.GetLiveManagersResponseDocument;
import hasthi.manager.ManageMeRequestDocument;
import hasthi.manager.ManagerAddress;
import hasthi.manager.ManagerContextDocument;
import hasthi.manager.ManagerPingRequestDocument;
import hasthi.manager.ManagerPingResponseDocument;
import hasthi.manager.AssignResourcesDocument.AssignResources;
import hasthi.manager.AssignResourcesDocument.AssignResources.ResourceData;
import hasthi.manager.CoordinatorStepDownNoticeDocument.CoordinatorStepDownNotice;
import hasthi.manager.ElectedNoticeDocument.ElectedNotice;
import hasthi.manager.GetLiveManagersResponseDocument.GetLiveManagersResponse;
import hasthi.manager.ManageMeRequestDocument.ManageMeRequest.ResourceDetails;
import hasthi.manager.ManagementRingDocument.ManagementRing;
import hasthi.manager.monitering.VerificationTools;

import java.io.IOException;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.Vector;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.xmlbeans.XmlCursor;
import org.apache.xmlbeans.XmlObject;
import org.xmlpull.v1.builder.XmlElement;

import xsul.http_server.HttpServerException;
import xsul.processor.DynamicInfosetProcessorException;
import xsul.processor.soap_over_http.SoapHttpDynamicInfosetProcessor;
import edu.indiana.extreme.wsdm.ShutDownRequestDocument;
import edu.indiana.extreme.wsdm.ShutDownRequestDocument.ShutDownRequest;

public class BootstrapNode{
    private static Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);
//    private PastryNode node;
    private int managerCount;
//    private Endpoint endpoint;
    private List<ManagerAddress> managers = new Vector<ManagerAddress>();
    private ManagerAddress coordinator;
    private BaseContext baseContext;
//    private ScribeImpl myScribe;
//    private Topic myTopic;
    
    private BootstrapMessageProcessor messageProcessor;
    
    private MulticastContianier multicastContainer;
    private UnicastContianier unicastContianier;
   // private SafeScheduler scheduler;
    
    private ManageMeMessageBag manageMeMessageBag;
    
    public BootstrapNode() throws HasthiException {
        this(new BaseContext());
        
    }
    
    public BootstrapNode(BaseContext baseContext) throws HasthiException {
        this.baseContext = baseContext;
        messageProcessor = new BootstrapMessageProcessor();
        multicastContainer = new P2PContainer(baseContext, messageProcessor,"127.0.0.1",
                baseContext.getBoostrapP2PPort(),baseContext.getBoostrapP2PPort());
//        scheduler = new SafeScheduler(40,new SafeScheduler.JobCancelledCallback() {
//            public void jobCancelled(Runnable runnable) {
//                log.warn("Job Cancelled "+runnable);
//            }
//        },false);
    }
    
    public void startNode() throws HasthiException{
        try {
            multicastContainer.startContainer();
            unicastContianier = ComponetFactory.createSoapContianier(baseContext, messageProcessor,baseContext.getBoostrapSoapPort(),false);
            manageMeMessageBag = new ManageMeMessageBag(unicastContianier);
//            HttpMiniServer server = new HttpMiniServer(9002);
//              setServer(server);
            unicastContianier.startContainer();
              System.out.println("bootstrapnode started on "+unicastContianier.getContainierAddress());
              //super.start();
        } catch (HttpServerException e) {
            throw new HasthiException(e,FaultCode.InitalizationError);
        } catch (DynamicInfosetProcessorException e) {
            throw new HasthiException(e,FaultCode.InitalizationError);
        } catch (HasthiException e) {
            throw new HasthiException(e,FaultCode.InitalizationError);
//        } catch (IOException e) {
//            throw new HasthiException(e,FaultCode.InitalizationError);
        }
    }
    
//    public void start(){
//        try {
//            Environment env = new Environment();
//            // disable the UPnP setting (in case you are testing this on a NATted
//            // LAN)
//            env.getParameters().setString("nat_search_policy", "never");
//
//            // the port to use locally
//            int bindport = 9001;
//
//            // build the bootaddress from the command line args
//            InetAddress bootaddr = InetAddress.getLocalHost();
//            int bootport = 9001;
//            InetSocketAddress bootaddress = new InetSocketAddress(bootaddr, bootport);
//
//            // Generate the NodeIds Randomly
//            NodeIdFactory nidFactory = new RandomNodeIdFactory(env);
//            
//            // construct the PastryNodeFactory, this is how we use
//            // rice.pastry.socket
//            PastryNodeFactory factory = new SocketPastryNodeFactory(nidFactory, bindport, env);
//
//            // This will return null if we there is no node at that location
//            NodeHandle bootHandle = ((SocketPastryNodeFactory) factory).getNodeHandle(bootaddress);
//
//            // construct a node, passing the null boothandle on the first loop will
//            // cause the node to start its own ring
//            node = factory.newNode(bootHandle);
//
//            // the node may require sending several messages to fully boot into the
//            // ring
//            synchronized (node) {
//                while (!node.isReady() && !node.joinFailed()) {
//                    // delay so we don't busy-wait
//                    node.wait(500);
//
//                    // abort if can't join
//                    if (node.joinFailed()) {
//                        throw new IOException("Could not join the FreePastry ring.  Reason:"
//                                + node.joinFailedReason());
//                    }
//                }
//            }
//            // We are only going to use one instance of this application on each PastryNode
//            endpoint = node.buildEndpoint(this, "myinstance");
//            // the rest of the initialization code could go here
//            endpoint.setDeserializer(new XmlMessageDeserializer());
//            // now we can receive messages
//            
//            /**
//             * Even though Boot strap node do not need broadcasts, we add it so
//             * root of the multicast tree does not fail with coordinator. Bootstrap node may also keep track 
//             * of the coordinator by listening to the broadcasts. 
//             */
//            myScribe = new ScribeImpl(node,"myScribeInstance");
////            // construct the topic
//            myTopic = new Topic(new PastryIdFactory(node.getEnvironment()), "admin");
//            // now we can receive messages
//            endpoint.register();
//            myScribe.subscribe(myTopic, this);
//            
//            System.out.println("Boostrap Node started as "+ node.getId());
//            
//            HttpMiniServer server = new HttpMiniServer(9002);
//            setServer(server);
//            super.start();
//            
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }

    
//    public void deliver(Id id, Message message) {
//        XmlMessage requestMessage = (XmlMessage)message;
//        String soapAction = (String)requestMessage.getAction();
//        String replyToAddress = requestMessage.getReplyTo();
//        
//        
//        
//        
//        try {
//            XmlObject requestBodyContent = requestMessage.getBodyContent();
//            
//            if(requestBodyContent instanceof GetCoordinatorDocument){
//                
//            }else {
//                XmlObject result = processMessageConcreate(requestBodyContent);
//                if(result != null && replyToAddress != null){
//                    rice.pastry.Id returnId = rice.pastry.Id.build(replyToAddress);
//                    XmlMessage responseMessage = new XmlMessage(result,node.getId(),soapAction+ "Response");
//                    RelatesToDocument relatesToDocument = RelatesToDocument.Factory.newInstance();
//                    relatesToDocument.addNewRelatesTo().setStringValue(requestMessage.getMessageID());
//                    responseMessage.addHeader(relatesToDocument);
//                    sendMessage(returnId, responseMessage);
//                }
//            }
//        } catch (HasthiException e) {
//            log.debug(e);
//            if(replyToAddress != null){
//                rice.pastry.Id returnId = rice.pastry.Id.build(replyToAddress);
//                XmlObject faultMessage = e.createFaultAsXmlBeans(true);
//                XmlMessage responseMessage = new XmlMessage(faultMessage,node.getId(),soapAction+ "Response");
//                RelatesToDocument relatesToDocument = RelatesToDocument.Factory.newInstance();
//                relatesToDocument.addNewRelatesTo().setStringValue(requestMessage.getMessageID());
//                responseMessage.addHeader(relatesToDocument);
//                sendMessage(returnId, responseMessage);
//            }
//        }
//    }
//    public boolean forward(RouteMessage message) {return true;}
//    public void update(rice.p2p.commonapi.NodeHandle handle, boolean joined) {}
//    
//    /**
//     * Send a message to a give recipeint
//     * @param recipient
//     * @param message
//     */
//    public void sendMessage(Id recipient, XmlMessage message){
//        log.debug("Send message  from "+ endpoint.getId() + " to "+ recipient );
//        ToDocument toDocument = ToDocument.Factory.newInstance();
//        toDocument.addNewTo().setStringValue(recipient.toStringFull());
//        message.addHeader(toDocument);
//        endpoint.route(recipient, message, null);
//    }
    
    
//    public boolean anycast(Topic topic, ScribeContent message) {
//        log.debug("Recived a Anycast at boostrap node") ;
//        try {
//            if(message instanceof AbstractXmlMesssage){
//                AbstractXmlMesssage requestMessage = (AbstractXmlMesssage)message;
//                String soapAction = (String)requestMessage.getAction();
//                String replyToAddress = requestMessage.getReplyTo();
//
//                XmlObject requestBodyContent = requestMessage.getBodyContent();
//                if(requestBodyContent instanceof GetCoordinatorDocument){
//                    synchronized (this) {
//                        if(managerCount == 0){
//                            managerCount++;
//                            GetCoordinatorResponseDocument result = GetCoordinatorResponseDocument.Factory.newInstance();
//                            result.addNewGetCoordinatorResponse().addNewAddress().setSoapAddress(ManagerConstants.SELF_ADDRESS);
//                            
//                            rice.pastry.Id returnId = rice.pastry.Id.build(replyToAddress);
//                            XmlMessage responseMessage = new XmlMessage(result,node.getId(),soapAction+ "Response");
//                            
//                            RelatesToDocument relatesToDocument = RelatesToDocument.Factory.newInstance();
//                            relatesToDocument.addNewRelatesTo().setStringValue(requestMessage.getMessageID());
//                            responseMessage.addHeader(relatesToDocument);
//
//                            ToDocument toDocument = ToDocument.Factory.newInstance();
//                            toDocument.addNewTo().setStringValue(returnId.toStringFull());
//                            responseMessage.addHeader(toDocument);
//                            endpoint.route(returnId, responseMessage, null);
//                            log.info("Give go ahead to become a manager to "+returnId);
//                            return false;
//                        }else{
//                            return true;
//                        }
//                    }   
//                }
//            }else{
//                throw new XmfException("Unknown Message type "+ message.getClass());
//            }
//            return true;
//        } catch (XmfException e) {
//            // TODO Auto-generated catch block
//            e.printStackTrace();
//            return true;
//        }
//    }
//    public void childAdded(Topic topic, rice.p2p.commonapi.NodeHandle child) {}
//    public void childRemoved(Topic topic, rice.p2p.commonapi.NodeHandle child) {}
//    public void deliver(Topic topic, ScribeContent content) {}
//    public void subscribeFailed(Topic topic) {}
    
//    @Override
//    public XmlElement processMessage(XmlElement request) throws DynamicInfosetProcessorException {
//        return null;
//    }
//    
//    public XmlDocument processHttpXml(final HttpServerRequest req, HttpServerResponse arg1,
//            XmlDocument soapDoc) throws DynamicInfosetProcessorException {
//        try {
//            XSULMessageContext context = new XSULMessageContext(soapDoc,req.getPath());
//            SoapUtil soapFragrance = context.getSoapFragrance();
//            XmlObject responseBodyContent = messageProcessor.processMessage(context);
//            return responseBodyContent == null? null : soapFragrance.wrapBodyContent(XBeansUtil.xmlObjectToXmlElement(responseBodyContent));
//        } catch (UnsupportedSoapVersion e) {
//            throw new DynamicInfosetProcessorException(e.getMessage(),e);
//        } catch (XsulException e) {
//            throw new DynamicInfosetProcessorException(e.getMessage(),e);
//        } catch (XmlBuilderException e) {
//            throw new DynamicInfosetProcessorException(e.getMessage(),e);
//        } catch (HasthiException e) {
//            throw new DynamicInfosetProcessorException(e.getMessage(),e);
//        }
//    }
    
    
    

    
    
//    public boolean anycast(Topic topic, ScribeContent content) {
//        // TODO Auto-generated method stub
//        return false;
//    }
//
//    public void childAdded(Topic topic, rice.p2p.commonapi.NodeHandle child) {
//        // TODO Auto-generated method stub
//        
//    }
//
//    public void childRemoved(Topic topic, rice.p2p.commonapi.NodeHandle child) {
//        // TODO Auto-generated method stub
//        
//    }
//
//    public void deliver(Topic topic, ScribeContent content) {
//        // TODO Auto-generated method stub
//        
//    }
//
//    public void subscribeFailed(Topic topic) {
//        // TODO Auto-generated method stub
//        
//    }
    
    
    

    public static void main(String[] args) throws Exception{
        if(args.length == 1 && "reset".equals(args[0])){
            UnicastContianier contianier;
            String bootstrapUrl;
            BaseContext baseContext = new BaseContext();
            contianier = ComponetFactory.createSoapContianier(baseContext, null);
            try {
                /** this is a tempeory hack, this message does not shutdown but reset the bootstrap node **/
                ShutDownRequestDocument requestDocument = ShutDownRequestDocument.Factory
                        .newInstance();
                ShutDownRequest shutDownRequest = requestDocument.addNewShutDownRequest();
                bootstrapUrl = "http://" + baseContext.getBoostrapHost() + ":"
                        + baseContext.getBoostrapSoapPort();
                MessageContext message = new SimpleMessageContext(requestDocument, bootstrapUrl,
                        WSDMConstants.ACTION_SHUTDOWN);
                contianier.send(message);
                System.out.println("Reset bootstrap "+ bootstrapUrl);
            }catch (Exception e) {
                e.printStackTrace();
            }finally{
                contianier.stopContainer();
                System.exit(0);
            }
        }else{
            BootstrapNode bootstrapNode = new BootstrapNode();
            bootstrapNode.startNode();
        }
    }
    
    public class BootstrapMessageProcessor implements MessageProcessor{
        public BootstrapMessageProcessor() throws HasthiException{
        }
        public void addContainer(CommunicationContianier container) throws HasthiException {
        }
        public XmlObject processMessage(MessageContext context)
                throws HasthiException {
            XmlObject requestBodyContent = context.getRequest();
            try {
                if(requestBodyContent == null){
                    String url = context.getSoapEndpoint();
                    if(url.contains(HasthiConstants.SYSTEM_STATUS_PREFIX)){
                        if(coordinator != null){
                            MessageContext messageContext = new SimpleMessageContext(null,coordinator.getSoapAddress()+"/"+HasthiConstants.SYSTEM_STATUS_PREFIX,
                                    HasthiConstants.MANAGER_PING_ACION);
                            return unicastContianier.sendReceive(messageContext);
                        }else{
                            return null;
                        }
                    }else{
                        ManagerContextDocument managerContextDocument = ManagerContextDocument.Factory.newInstance();
                        if(coordinator != null){
                            ManagerPingRequestDocument requestDocument = ManagerPingRequestDocument.Factory.newInstance();
                            requestDocument.addNewManagerPingRequest();
                            
                            MessageContext messageContext = new SimpleMessageContext(requestDocument,coordinator.getSoapAddress(),
                                    HasthiConstants.MANAGER_PING_ACION);
                            ManagerPingResponseDocument responseDocument = (ManagerPingResponseDocument)unicastContianier.sendReceive(messageContext);
                            
                            managerContextDocument.setManagerContext(responseDocument.getManagerPingResponse().getManagerContext());
                            
                        }else{
                            ManagementRing managementRing = managerContextDocument.addNewManagerContext().addNewManagementRing();
                            for(ManagerAddress managerAddress:managers){
                                managementRing.addNewManagerData().setManagerAddress(managerAddress);
                            }
                        }
                        XmlCursor c = managerContextDocument.newCursor();
                        c.toStartDoc();
                        c.toFirstChild();
                        c.insertProcInst("xml-stylesheet", "type=\"text/xsl\" href=\"http://www.cs.indiana.edu/~hperera/dist/system.xslt\"");
                        c.dispose();
                        return managerContextDocument;
                    }
                }else if(requestBodyContent instanceof GetCoordinatorDocument){
//                    if(requestPath != null){
                        //then this is SOAP
                        GetCoordinatorDocument request = (GetCoordinatorDocument)requestBodyContent;
                        ManagerAddress requesterAddress = request.getGetCoordinator().getRequesterAddress();
                        GetCoordinatorResponseDocument response = GetCoordinatorResponseDocument.Factory.newInstance();

                        if(requesterAddress != null){
                            managers.add(requesterAddress);
                            synchronized (this) {
                                if(managerCount == 0){
                                    managerCount++;
                                    coordinator = requesterAddress;
                                    response.addNewGetCoordinatorResponse().addNewAddress().setSoapAddress(HasthiConstants.SELF_ADDRESS);
                                }else{
                                    response.addNewGetCoordinatorResponse().setAddress(coordinator);
                                }
                            }

                        }else{
                            //requester is not a manager, so return with the current coordiantor
                            response.addNewGetCoordinatorResponse().setAddress(coordinator);
                        }
                        return response;
//                    }else{
//                      //If it is a get coordinator, this will foward it to a random manager and return the results
//                        ManagerAddress managerAddress = managers.get((int)(System.currentTimeMillis()%managers.size()));
//                        rice.pastry.Id fowardId = rice.pastry.Id.build(managerAddress.getP2Paddress());
//                        XmlMessage response = container.sendReciveMessage(fowardId, new XmlMessage(requestBodyContent,container.getNode().getId(),soapAction));
//                        return response.getBodyContent();
//                    }
                }else if(requestBodyContent instanceof GetLiveManagersDocument){
                    GetLiveManagersResponseDocument responseDocument = GetLiveManagersResponseDocument.Factory.newInstance();
                    GetLiveManagersResponse response = responseDocument.addNewGetLiveManagersResponse();
                    for(int i = 0;i<Math.min(5, managers.size());i++){
                        ManagerAddress returnaddress = response.addNewManagerAddress();
                        ManagerAddress address = managers.get(i);
                        returnaddress.setSoapAddress(address.getSoapAddress());
                        returnaddress.setP2Paddress(address.getP2Paddress());
                    }
                    return responseDocument;
                }else if(requestBodyContent instanceof ShutDownRequestDocument){
                    managerCount = 0;
                    managers.clear();
                    coordinator = null;
                }else if(requestBodyContent instanceof ManageMeRequestDocument){
                    //scheduler.submit(new SendMessageAction(context));
                    ManageMeRequestDocument requestDocument = (ManageMeRequestDocument)requestBodyContent;
                    hasthi.manager.ManageMeRequestDocument.ManageMeRequest manageMeRequest = requestDocument.getManageMeRequest();
                    
                    ResourceDetails[] resourceDetails = manageMeRequest.getResourceDetailsArray();
                    for(ResourceDetails resourceDetail:resourceDetails ){
                        String managementEpr = resourceDetail.getManagmentAgentEpr();
                        if(managementEpr.contains("localhost") || managementEpr.contains("127.0.")){
                            log.warn("Ignore ill formartted management messages for  "+ resourceDetail.getResourceName() + "@" + managementEpr);
                        }else{
                            log.info("resource "+ resourceDetail.getResourceName() +"("+ resourceDetail.getManagmentAgentEpr()+") join message at bootstrap node");
                            if(resourceDetail != null){
                                manageMeMessageBag.add(new ManageMeRequest(resourceDetail.getResourceName(),managementEpr));    
                            }else{
                                log.error("resoved a null resource detailed in a manage me message");
                            }
                                
                        }
                    }
                }else if(requestBodyContent instanceof ElectedNoticeDocument){
                    ElectedNotice electedNotice = ((ElectedNoticeDocument)requestBodyContent).getElectedNotice();
                    coordinator = electedNotice.getAddress();
                }else if(requestBodyContent instanceof CoordinatorStepDownNoticeDocument){ 
                    CoordinatorStepDownNotice coordinatorStepDownNotice = 
                        ((CoordinatorStepDownNoticeDocument)requestBodyContent).getCoordinatorStepDownNotice();
                    coordinator = coordinatorStepDownNotice.getAppointedCoordinator(); 
                }else if(requestBodyContent instanceof  CoordinatorHeartbeatDocument)  {    
                    coordinator = ((CoordinatorHeartbeatDocument)requestBodyContent).getCoordinatorHeartbeat().getAddress();
                }else{
                    throw new HasthiException("Unkonwn Message "+ requestBodyContent.getClass() 
                            + " with "+ XmlBeanUtils.prettyPrint(requestBodyContent),FaultCode.InvaliedRequest);
                }
                return null;
            } catch (Exception e) {
                throw new HasthiException(e,FaultCode.InternalServiceError);
            }
        }
    }
    
    
    public class SendMessageAction implements Runnable{
        private MessageContext context;
        public SendMessageAction(MessageContext context){
            this.context = context;
        }
        public void run() {
          //If it is manage Me request, forward it
            try {
                if(coordinator != null){
                    try {
                        context.setSoapEndpoint(coordinator.getSoapAddress());
                        unicastContianier.send(context);
                        return;
                    } catch (Throwable e) {
                        coordinator = null;
                        log.error(e);
                    }
                }
                if(managers.size() >1){
                    ManagerAddress address;
                    synchronized (managers) {
                        int index = Utils.randomInt(managers.size());
                        address = managers.get(index);
                    }
                    context.setSoapEndpoint(address.getSoapAddress());
                    unicastContianier.sendReceive(context);
                }else{
                    log.warn("No Manager found in bootstrap Node");
                }
            } catch (HasthiException e) {
                log.warn("Failed to foward "+ context.getRequest() 
                        + " "+ e.getMessage());
                ErrorManager.error(e);
            }
        }
    }
    
    
    public class ManageMeMessageBag extends AbstractMessageBag<ManageMeRequest>{
        public ManageMeMessageBag(UnicastContianier soapContainer) {
            super(soapContainer);
        }

        @Override
        protected void processBatch(Set<ManageMeRequest> list) throws HasthiException {
            if (list.size() == 0 || coordinator == null) {
                return;
            }

            ManageMeRequestDocument manageMeRequestDocument = ManageMeRequestDocument.Factory.newInstance();
            hasthi.manager.ManageMeRequestDocument.ManageMeRequest manageMeRequests = manageMeRequestDocument.addNewManageMeRequest();

            
            for (ManageMeRequest request : list) {
                ResourceDetails resourceDetail = manageMeRequests.addNewResourceDetails();
                resourceDetail.setManagmentAgentEpr(request.managementAgentUrl);
                resourceDetail.setResourceName(request.name);

            }
            MessageContext requestMessage = new SimpleMessageContext(manageMeRequestDocument, 
                    coordinator.getSoapAddress(), MANAGE_ME_ACTION);
            soapContainer.sendReceive(requestMessage);
        }
    } 
    
    
//
//    @Override
//    public void start() throws IOException, DynamicInfosetProcessorException {
//        throw new DynamicInfosetProcessorException("Use startNode() instead");
//    }
}

